package net.bwie.realtime.jtp.dwd.trade.function;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import net.bwie.realtime.jtp.common.utils.HbaseUtil;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;

import java.util.Collections;
import java.util.concurrent.*;
import java.util.function.Supplier;

public class RequestDicDimAsyncFunction extends RichAsyncFunction<String, String> {
   private ExecutorService executorService;
   @Override
   public void open(Configuration parameters) throws Exception {
       executorService= Executors.newScheduledThreadPool(10);
   }
    @Override
    public void asyncInvoke(String input, ResultFuture<String> resultFuture) throws Exception {
        JSONObject jsonObject = JSON.parseObject(input);
        String rowKey = jsonObject.getString("activity_type");
        final Future<String> result = executorService.submit(
                new Callable<String>() {

                    @Override
                    public String call() throws Exception {
                        return HbaseUtil.getData("dim_base_dic", rowKey, "info", "dic_name");
                    }
                }
        );
        CompletableFuture.supplyAsync(new Supplier<String>() {
                    @Override
                    public String get() {
                        try {
                            String requestValue = result.get();
                            // 将请求结果放入json中
                            jsonObject.put("source_type_name", requestValue);
                        }catch (InterruptedException | ExecutionException e) {
                            // 请求失败exception
                            jsonObject.put("source_type_name", "exception");
                        }
                        return jsonObject.toString();
                    }
                }
        ).thenAccept((String dbResult)->
        {
            resultFuture.complete(Collections.singleton(dbResult));
        });
    }

    @Override
    public void timeout(String input, ResultFuture<String> resultFuture) throws Exception {
        JSONObject jsonObject = JSON.parseObject(input);
        jsonObject.put("source_type_name", "timeout");
        resultFuture.complete(Collections.singleton(jsonObject.toString()));
    }
    @Override
    public void close() throws Exception {
        executorService.shutdown();
    }
}
